package red.ouo.distributedtools.redis.message.core;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.Assert;
import red.ouo.distributedtools.standard.callback.Callback;
import red.ouo.distributedtools.standard.callback.CallbackStatus;
import red.ouo.distributedtools.standard.message.MessageBody;
import red.ouo.distributedtools.standard.message.MessageCoreInterface;
import red.ouo.distributedtools.standard.message.MessageStatus;
import red.ouo.distributedtools.standard.message.QueueEntity;
import red.ouo.distributedtools.redis.message.config.RedisMessageConfig;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

public class RedisMessageCore<T> implements MessageCoreInterface<T> {
    public int MAX_AGE = 5;

    private Map<String, ThreadPoolExecutor> map = new HashMap<>();
    private RedisTemplate<String, Object> redisTemplate;
    public Options options;
    public RedisMessageConfig config;
    public String queueName;
    public static String queuePrefix = "KELE_QUEUE_";

    private RedisMessageCore(RedisTemplate redisTemplate, RedisMessageConfig config) {
        this.redisTemplate = redisTemplate;
        this.config = config;
    }

    @Override
    public MessageBody<T> send(MessageBody<T> messageBody) {
        sendBase(messageBody);
        return messageBody;
    }

    public MessageBody<T> sendBase(MessageBody<T> messageBody) {
        QueueEntity e = this.config.getExecutorService(getRelativeQueueName());
        Assert.notNull(e.getExecutorService(), String.format("The queue name [%s] is not allowed! Please go to the configuration center to register!", getRelativeQueueName()));
        redisTemplate.execute((RedisCallback<String>) redisConnection -> redisConnection.lPush(getRelativeQueueName().getBytes(), JSON.toJSONBytes(messageBody)) == null ? null : "OK");
        return messageBody;
    }

    public final T parseObject(String data) {
        Type[] params = ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments();
        return JSONObject.parseObject(data, params[0]);
    }

    public static RedisMessageCore.Builder connect(RedisTemplate redisTemplate, RedisMessageConfig config) {
        return (new RedisMessageCore(redisTemplate, config).new Builder(redisTemplate));
    }

    /**
     * 阻塞拉取一条
     *
     * @param callback
     * @return
     */
    public Object fetchUnitaryTimeOut(Callback<MessageBody<T>> callback, Integer timeOut) {
        return fetchSomeTimeOut((e,t) -> {
            callback.ok(e,t);
            return true;
        }, 1, timeOut);
    }

    @Override
    public Object fetchSomeTimeOut(CallbackStatus<MessageBody<T>> callback, Integer count, Integer timeOut) {
        QueueEntity e = config.getExecutorService(getRelativeQueueName());
        Assert.isTrue(e!=null&&e.getExecutorService()!=null, String.format("The queue name [%s] is not allowed! Please go to the configuration center to register!", queueName));

        return redisTemplate.execute((RedisCallback<String>) redisConnection -> {
            e.getExecutorService().submit(() -> {
                for (int i = 0; count <= 0 ? true : i < count; i++) {
                    List<byte[]> end = redisConnection.bLPop(timeOut, getRelativeQueueName().getBytes());
                    if (end != null && end.size() >= 2) {
                        MessageBody<T> temp = JSONObject.parseObject(new String(end.get(1)), MessageBody.class);
                        boolean keep = callback.ok(temp,this);
                        if (!keep) break;
                    }
                }
            });
            return "OK";
        });
    }


    @Override
    public void operation(MessageBody<T> messageBody, MessageStatus messageStatus) {
        switch (messageStatus.code) {
            case 200:
            case 400:
                //无任何操作
                break;
            case 500:
                //重试
                int age = messageBody.getAge();
                if(age<=MAX_AGE){
                    messageBody.setAge((byte) (messageBody.getAge()+1));
                    send(messageBody);
                }else {
                    throw new RuntimeException("消息重试次数超限");
                }
                break;
        }
    }

    /**
     * 最终处理结果确认
     *
     * @param callback
     * @return
     */
    public Object fetchUnitary(Callback<MessageBody<T>> callback) {
        return fetchUnitaryTimeOut(callback, 0);
    }

    /**
     * 消息队列构造参数
     */
    public class Builder {
        RedisTemplate redisTemplate;

        public ThreadPoolExecutor registerQueue(String queueName, ThreadPoolExecutor runWith) {
            return map.put(getRelativeQueueName(queueName), runWith);
        }

        public Builder(RedisTemplate redisTemplate) {
            this.redisTemplate = redisTemplate;
        }

        public Builder queue(String queueName) {
            RedisMessageCore.this.queueName = queueName;
            return this;
        }

        public Builder queuePrefix(String queuePrefix) {
            RedisMessageCore.this.queuePrefix = queuePrefix;
            return this;
        }

        public Options options() {
            return options = new Options(RedisMessageCore.this);
        }

    }

    /**
     * 消息队列 可选配置
     *
     * @param <T>
     */
    public class Options<T> {
        RedisMessageCore<T> messageCore;

        public Options(RedisMessageCore messageCore) {
            this.messageCore = messageCore;
        }
        public Options<T> setMaxAge(Byte max){
            messageCore.MAX_AGE = max;
            return this;
        }

        public RedisMessageCore<T> build() {
            return messageCore;
        }
    }

    public String getRelativeQueueName(){
        return this.queuePrefix+this.queueName;
    }
    public String getRelativeQueueName(String queueName){
        return queuePrefix+queueName;
    }
}
